package org.bigwinner.flinkLearning.Functions;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.bigwinner.flinkLearning.Message.MysqlMessage;
import org.bigwinner.flinkLearning.Sinks.MySqlTwoPhaseCommitSink;
import org.bigwinner.flinkLearning.utils.CastTypeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author: IT大狮兄
 * @date: 2021/8/1 上午10:59
 * @version:
 * @description:
 */
public class String2MysqlMessagePF<T, T1> extends ProcessFunction<String, MysqlMessage> {
    private static final Logger LOGGER = LoggerFactory.getLogger(String2MysqlMessagePF.class);
    int count = 0;
    @Override
    public void processElement(String record, Context context, Collector<MysqlMessage> collector) throws Exception {
        count++;
        LOGGER.warn("count: {}, msg: {}", count, record);
        Map dataMap = JSON.parseObject(record, Map.class);
        MysqlMessage mysqlMessage = new MysqlMessage();
        mysqlMessage.setIcid(CastTypeUtils.castString(dataMap.get("icid"), ""));
        mysqlMessage.setTelnum(CastTypeUtils.castString(dataMap.get("telnum"), ""));
        // 模拟接收处理第10条数据的时候发生bug
        if (count == 10) throw new Exception("bug......");
        collector.collect(mysqlMessage);


    }
}
